package org.apache.flink.runtime.resourceconsumer;

import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import org.apache.flink.runtime.allocatedeploy.execution.ExecutionID;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.resourceprovider.ResourceConsumptionReport;

/**
 * Default {@link ExecutionDeploymentReconciler} implementation. Detects missing/unknown
 * deployments, and defers to a provided {@link ExecutionDeploymentReconciliationHandler} to resolve
 * them.
 */
public class DefaultExecutionDeploymentReconciler implements ExecutionDeploymentReconciler {

    private final ExecutionDeploymentReconciliationHandler handler;

    public DefaultExecutionDeploymentReconciler(ExecutionDeploymentReconciliationHandler handler) {
        this.handler = handler;
    }

    @Override
    public void reconcileExecutionDeployments(
            ResourceID taskExecutorHost,
            ResourceConsumptionReport executionDeploymentReport,
            Map<ExecutionID, ExecutionDeploymentState> expectedDeployedExecutions) {
        final Set<ExecutionID> unknownExecutions =
                new HashSet<>(executionDeploymentReport.getConsumptions());
        final Set<ExecutionID> missingExecutions = new HashSet<>();

        for (Map.Entry<ExecutionID, ExecutionDeploymentState> execution :
                expectedDeployedExecutions.entrySet()) {
            boolean deployed = unknownExecutions.remove(execution.getKey());
            if (!deployed && execution.getValue() != ExecutionDeploymentState.PENDING) {
                missingExecutions.add(execution.getKey());
            }
        }
        if (!unknownExecutions.isEmpty()) {
            handler.onUnknownDeploymentsOf(unknownExecutions, taskExecutorHost);
        }
        if (!missingExecutions.isEmpty()) {
            handler.onMissingDeploymentsOf(missingExecutions, taskExecutorHost);
        }
    }
}
